package net.wlm.realtime.job;

import net.wlm.realtime.funcation.DimDataFilterFunction;
import net.wlm.realtime.funcation.DimHbaseSinkFunction;
import net.wlm.realtime.funcation.HbaseDimMapFunction;
import net.wlm.realtime.utils.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.HashMap;
import java.util.HashSet;

/**
 * * 从Kafka队列topic-db实时获取维度表数据，将其同步存储HBase维表中，其中insert/update类型数据插入更新，delete类型数据删除
 * *      第1. 多张维表；第2、多种类型数据
 */

public class RealtimeAsyncDataDimJob {
    public static void main(String[] args) throws Exception {
        // 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 数据源-source
        DataStream<String> stream = KafkaUtil.consumerKafka(env, "topic-db");
        // 数据转换-transformation
        DataStream<String> dimStream = handle(stream);
        // 数据接收器-sink
        dimStream.addSink(new DimHbaseSinkFunction());
        // 触发执行-execute
        env.execute();
    }
    /**
     * 按照表名称过滤获取对应维表数据流，保存HBase表，列簇统一为：info，指定字段名称
     */
    private static DataStream<String> handle(DataStream<String> stream) {
        // 维度表与主键映射
        HashMap<String, String> dimMap = new HashMap<String, String>() {{
            put("base_dic","dic_code");put("base_region","id");put("base_province","id"); put("base_trademark","id");
            put("base_category1","id");put("base_category2","id");put("base_category3","id");
            put("coupon_info","id");put("sku_info","id");put("user_info","id");
            put("activity_info","id");put("activity_rule","id");put("spu_info","id");
        }};
        // 维度表名称
        HashSet<String> dimSet = new HashSet<>(dimMap.keySet());
        // 过滤维表数据
        DataStream<String> syncStream = stream.filter(new DimDataFilterFunction(dimSet));
        // 添加字段值
        DataStream<String> hbaseStream = syncStream.map(new HbaseDimMapFunction(dimMap));
        return hbaseStream;
    }
}

























































